import asyncio
import io
import os
import re
import tempfile
import uuid
from enum import Enum
from io import BytesIO
from typing import List, Optional, Dict, Union, Type
from urllib.parse import quote

import pandas as pd
from fastapi import UploadFile, HTTPException
from fastapi.responses import StreamingResponse
from openpyxl.styles import Font, Alignment, Border, Side, PatternFill
from openpyxl.utils import get_column_letter
from pydantic_core import PydanticUndefined

from app.core.generic_types import ExportSchemaType, CreateSchemaType

def _convert_data_to_dataframe(data:List[ExportSchemaType], exclude_fields:Optional[List[str]]=None) -> pd.DataFrame:
    """
    将模型字段转换成数据DataFrame
    :param data: 导出数据列表
    :param exclude_fields: 要忽略的字段
    :return: DataFrame
    """
    exclude_field_list=["is_delete"]
    if exclude_fields is not None and len(exclude_fields) > 0:
        exclude_field_list.extend(exclude_fields)

    if not data or len(data) == 0:
        raise ValueError("源数据不能为空")

    model_fields = data[0].model_fields
    rows=[]
    for item in data:
        item_dict ={}
        for field_name,field_info  in model_fields.items():
            if field_name in exclude_field_list:
                continue
            value = getattr(item, field_name)
            if isinstance(value, Enum):
                value = value.description if hasattr(value, 'description') else str(value)
            if isinstance(value, bool):
                value = "是" if value else "否"
            name = field_info.description if hasattr(field_info, 'description') else field_name
            item_dict[name] = value
        rows.append(item_dict)

    # 创建DataFrame
    df = pd.DataFrame(rows)

    # 添加序号列
    df.insert(0, '序号', pd.Series(range(1, len(df) + 1)))
    return df

def _convert_model_to_dataframe(model:Type[ExportSchemaType], exclude_fields:Optional[List[str]]=None) -> pd.DataFrame:
    """
    将模型字段转换成数据DataFrame
    :param model: 导出模版的模型
    :param exclude_fields: 要忽略的字段
    :return: DataFrame
    """
    exclude_field_list=["is_delete","password","confirm_password","is_enabled"]
    if exclude_fields is not None and len(exclude_fields) > 0:
        exclude_field_list.extend(exclude_fields)

    model_fields = model.model_fields
    rows = []
    item_dict = {}

    # 收集字段信息（字段名和是否必填）
    for field_name, field_info in model_fields.items():
        if field_name in exclude_field_list:
            continue

        # 获取字段显示名
        display_name = field_info.description if hasattr(field_info, 'description') else field_name
        # 检查是否必填字段
        is_required = field_info.is_required() if hasattr(field_info, 'is_required') else False
        # 准备示例数据
        value = field_info.default if field_info.default is not None and field_info.default != PydanticUndefined else None

        if isinstance(value, Enum):
            value = value.description if hasattr(value, 'description') else str(value)
        if isinstance(value, bool):
            value = "是" if value else "否"

        # 约定带*号的为必填项
        item_dict[display_name + ("*" if is_required else "")] = value

    item_dict["备注"] = '带 * 的为必填项'

    rows.append(item_dict)

    # 创建DataFrame
    df = pd.DataFrame(rows)

    return df

def _apply_worksheet_styles(worksheet, df):
    """
    定义样式
    :param worksheet: 工作表
    :param df: 数据DataFrame
    """
    header_font = Font(bold=True, color="FFFFFF")
    header_fill = PatternFill("solid", fgColor="4F81BD")
    header_alignment = Alignment(horizontal="center", vertical="center")
    thin_border = Border(
        left=Side(style='thin'),
        right=Side(style='thin'),
        top=Side(style='thin'),
        bottom=Side(style='thin')
    )
    cell_alignment = Alignment(horizontal="center", vertical="center")

    # 设置表头样式
    for col in range(1, len(df.columns) + 1):
        cell = worksheet.cell(row=1, column=col)
        cell.font = header_font
        cell.fill = header_fill
        cell.alignment = header_alignment
        cell.border = thin_border
        if cell.value.endswith("*"):
            # 约定带*号的为必填项，将必填项字体设置为红色
            cell.font = Font(bold=True, color="FF0000", name=cell.font.name, size=cell.font.size)

    # 设置数据行样式
    for row in worksheet.iter_rows(min_row=2, max_row=len(df) + 1):
        for cell in row:
            cell.alignment = cell_alignment
            cell.border = thin_border

    # 调整列宽
    dims = {}
    for row in worksheet.iter_rows():
        for cell in row:
            if cell.value:
                cell_len = 0.7 * len(re.findall('([\u4e00-\u9fa5])', str(cell.value))) + len(str(cell.value))
                dims[cell.column] = max((dims.get(cell.column, 0), cell_len))

    for col, value in dims.items():
        # 设置列宽 get_column_letter 用于获取数字列号对应的字母列号，最后值 +5 是用来调整最终效果的
        worksheet.column_dimensions[get_column_letter(col)].width = value + 5

    # 设置表头行高
    worksheet.row_dimensions[1].height = 30

async def _generate_excel_response(df: pd.DataFrame,export_name: str,sheet_name: str,style_func: Optional[callable] = None) -> StreamingResponse:
    """
    生成Excel响应公共函数
    :param df: 数据DataFrame
    :param export_name: 导出文件名
    :param sheet_name: 工作表名
    :param style_func: 样式处理函数
    :return: StreamingResponse
    """
    loop = asyncio.get_running_loop()

    def sync_generate(_df: pd.DataFrame,_sheet_name: str,_style_func: Optional[callable] = None):
        file = BytesIO()
        temp_path = f"{tempfile.gettempdir()}/{uuid.uuid4()}.xlsx"

        try:
            with pd.ExcelWriter(path=temp_path,engine='openpyxl',date_format='YYYY-MM-DD',datetime_format='YYYY-MM-DD HH:MM:SS') as writer:
                # 写入数据
                _df.to_excel(writer,sheet_name=_sheet_name,index=False,header=[str(col) for col in _df.columns])

                # 应用样式
                if _style_func:
                    workbook = writer.book
                    worksheet = writer.sheets[sheet_name]
                    _style_func(worksheet, _df)

                # 保存到内存缓冲区
                workbook.save(file)

            file.seek(0)
            return file
        finally:
            if os.path.exists(temp_path):
                os.remove(temp_path)

    excel_file = await loop.run_in_executor(None, sync_generate, df, sheet_name, style_func)

    # 处理文件名编码
    filename = f"{export_name}.xlsx"
    try:
        filename.encode('latin-1')
    except UnicodeEncodeError:
        filename = quote(filename)

    headers = {
        "Content-Disposition": f"attachment; filename={filename}",
        "Content-Type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    }

    return StreamingResponse(
        content=excel_file,
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers=headers
    )

async def export_excel_template(model: Type[CreateSchemaType],export_name: str = "默认模版",sheet_name: str = "Sheet1",exclude_fields: Optional[list[str]] = None) -> StreamingResponse:
    """
    导出Excel模板，*结尾为必填字段
    :param model: 要导出的模版模型
    :param export_name: 导出的文件名称
    :param sheet_name: 工作表名
    :param exclude_fields: 要忽略字段
    :return: StreamingResponse
    """
    df = _convert_model_to_dataframe(model, exclude_fields=exclude_fields)

    return await _generate_excel_response(
        df=df,
        export_name=export_name,
        sheet_name=sheet_name,
        style_func=_apply_worksheet_styles
    )

async def export_excel(data: List[ExportSchemaType], export_name: str = "默认", sheet_name: str = "Sheet1", exclude_fields:Optional[list[str]]=None) -> StreamingResponse:
    """
    使用pandas将数据列表导出为Excel文件
    :param data: 要导出的数据列表
    :param export_name: 导出的文件名称
    :param sheet_name: 工作表名
    :param exclude_fields: 要忽略字段
    :return: StreamingResponse
    """
    df = _convert_data_to_dataframe(data, exclude_fields)
    return await _generate_excel_response(
        df=df,
        export_name=export_name,
        sheet_name=sheet_name,
        style_func=_apply_worksheet_styles
    )

async def read_excel(file:UploadFile,sheet_name: str = "Sheet1",skip_rows: int = 0)  -> List[Dict[str, Union[str, int, float, bool]]]:
    """
    读取excel文件，并转换成指定类型的数据
    :param file: 上传的excel文件
    :param sheet_name: 工作表名
    :param skip_rows: 开头跳过多少行
    :return: List[Dict[str, Union[str, int, float, bool]]]
    """
    loop = asyncio.get_running_loop()
    try:
        # 读取文件内容到内存
        contents = await file.read()

        def sync_read(_sheet_name: str,_skip_rows:int):
            # 使用pandas读取Excel
            # dtype=str 统一按字符串读取，避免类型推断问题
            # keep_default_na=False 不将空字符串转为NaN
            df = pd.read_excel(io.BytesIO(contents),sheet_name=_sheet_name,skiprows=skip_rows,dtype=str,keep_default_na=False)

            # 处理表头，分离字段名和必填标记
            original_columns = list(df.columns)
            required_columns = []
            cleaned_columns = []
            for col in original_columns:
                if isinstance(col, str) and col.endswith('*'):
                    required_columns.append(col.rstrip('*'))
                    cleaned_columns.append(col.rstrip('*'))
                else:
                    cleaned_columns.append(col)

            # 更新DataFrame列名
            df.columns = cleaned_columns

            # 检查必填字段是否有空值
            for col in required_columns:
                if col in df.columns and df[col].isnull().any():
                    raise HTTPException(
                        status_code=400,
                        detail=f"必填字段 '{col}' 存在空值"
                    )

            # 替换空字符串为None
            df = df.replace('', None)

            # 转换为字典列表
            result = df.to_dict('records')

            # 处理布尔值字段（将"是"/"否"转为True/False）
            for record in result:
                for key, value in record.items():
                    if isinstance(value, str):
                        if value.strip().lower() in ('是', 'yes', 'true', 't'):
                            record[key] = True
                        elif value.strip().lower() in ('否', 'no', 'false', 'f'):
                            record[key] = False
                        elif value.strip().lower() in ('男', '男性'):
                            record[key] = 0
                        elif value.strip().lower() in ('女', '女性'):
                            record[key] = 1
                        # 其他逻辑

            return result

        return await loop.run_in_executor(None, sync_read,sheet_name,skip_rows)

    except HTTPException:
        raise  # 直接抛出已有的HTTP异常
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail=f"解析Excel文件失败: {str(e)}"
        )

async def read_excel_to_model_list(model:Type[CreateSchemaType], file: UploadFile, sheet_name: str = "Sheet1", skip_rows: int = 0) -> List[CreateSchemaType]:
    """
    读取excel文件，并转换成指定类型的数据
    :param model: 返回的模型类型
    :param file: 上传的excel文件
    :param sheet_name: 工作表名
    :param skip_rows: 开头跳过多少行
    :return: List[CreateSchemaType]
    """
    data = await read_excel(file,sheet_name=sheet_name,skip_rows=skip_rows)

    model_fields = model.model_fields
    item_map = {}
    # 收集字段信息（字段名和是否必填）
    for field_name, field_info in model_fields.items():
        # 获取字段显示名
        display_name = field_info.description if hasattr(field_info, 'description') else field_name
        item_map[display_name] = field_name

    model_lst = []
    for item in data:
        item_dict={}
        for field_name, field_value in item.items():
            if field_name in item_map and field_value is not None:
                item_dict[item_map[field_name]] = field_value
        model_item = model(**item_dict)
        model_lst.append(model_item)
    return model_lst






